import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis
import java.util

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._

import scala.collection.JavaConversions._
import scala.collection.mutable


/**
  * Created by danke on 2020/3/31.
  */
object StreamingRedisOffset {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("networkWordCount")

    val ssc:StreamingContext = new StreamingContext(conf,Seconds(5))
    val topics = Array("spark-kafka")
    val groupId = "sparkaGroup"

    val kafkaParams = mutable.HashMap[String,Object](
      "bootstrap.servers"-> "hadoop01:9092",
      "key.deserializer"->classOf[StringDeserializer].getName,      //key的序列化器
      "value.deserializer"->classOf[StringDeserializer].getName,    //value的序列化器
      "group.id"-> groupId,
      "auto.offset.reset"->"latest",
      "enable.auto.commit"->"false"
    )

    //从redis中获取偏移量
    val offsetsMap = mutable.HashMap[TopicPartition,Long]()
    val jedis:Jedis = JedisUtils("192.168.132.100",6379)
    val partandOffset: util.Map[String,String] = jedis.hgetAll(topics(0) + "-" + groupId)
    for(part <- partandOffset){
      offsetsMap += (new TopicPartition(topics(0),part._1.toInt) -> part._2.toLong)
    }

    var kafkaDs:InputDStream[ConsumerRecord[String,String]] = KafkaUtils.createDirectStream(ssc,LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](topics,kafkaParams,offsetsMap))
    kafkaDs.foreachRDD(rdd=>{
      //判断rdd非空
      if(!rdd.isEmpty()){
        //获取每一个分区的消费的偏移量
        val ranges:Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //ranges.foreach(println)
        //业务逻辑
        rdd.foreach(t =>{
          println(t.value())
         // ,t.offset(),t.partition()
        })

        //偏移量，写入到redis中
      //  val curTopic = topics(0)
        for(t <- ranges){
          val jedis:Jedis = JedisUtils("192.168.132.100",6379)
          jedis.hset(t.topic + "-" + groupId,t.partition + "",t.untilOffset +"")
          jedis.close()
        }
      }
    })
    ssc.start()
    ssc.awaitTermination()
  }
}
